{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "33f6332e",
   "metadata": {},
   "source": [
    "# 多动作多智能体\n",
    "\n",
    "在上一章中，我们简要讨论了单智能体的创建。虽然对许多情况来说，单智能体可能已经足够，但更复杂的任务通常需要协作和团队合作，这也就是多智能体为什么必不可少的原因。MetaGPT的核心优势也在于轻松灵活地开发一个智能体团队。在MetaGPT框架下，用户可以通过少量代码实现智能体之间的交互。\n",
    "\n",
    "完成本节，你将能够：\n",
    "\n",
    "1. 理解智能体之间如何进行交互\n",
    "2. 开发你的第一个智能体团队\n",
    "\n",
    "\n",
    "## **开发你的第一个智能体团队**\n",
    "\n",
    "希望你会发现软件创业示例很有启发。也许现在你已经有了灵感，想要开发一个根据你的独特需求而定制的智能体团队。在本节中，我们将继续在[智能体入门](https://docs.deepwisdom.ai/main/zh/guide/tutorials/agent_101.html)中的简单代码示例中添加更多角色，并引入智能体之间的交互协作。\n",
    "\n",
    "让我们还雇佣一名测试人员和一名审阅人员携手与编码人员一起工作。这开始看起来像一个开发团队了，不是吗？总的来说，我们需要三个步骤来建立团队并使其运作：\n",
    "\n",
    "1. 定义每个角色能够执行的预期动作\n",
    "2. 基于标准作业程序（SOP）确保每个角色遵守它。通过使每个角色观察上游的相应输出结果，并为下游发布自己的输出结果，可以实现这一点。\n",
    "3. 初始化所有角色，创建一个带有环境的智能体团队，并使它们之间能够进行交互。\n",
    "\n",
    "完整的代码在本教程的末尾可用\n",
    "\n",
    "**定义动作和角色**\n",
    "\n",
    "与前面课程相同的过程，我们可以定义三个具有各自动作的Role：\n",
    "\n",
    "1. SimpleCoder具有 SimpleWriteCode 动作，接收用户的指令并编写主要代码\n",
    "2. SimpleTester 具有 SimpleWriteTest 动作，从 SimpleWriteCode 的输出中获取主代码并为其提供测试套件\n",
    "3. SimpleReviewer 具有 SimpleWriteReview 动作，审查来自 SimpleWriteTest 输出的测试用例，并检查其覆盖范围和质量\n",
    "\n",
    "通过上述概述，我们使得 SOP（标准作业程序）变得更加清晰明了。接下来，我们将详细讨论如何根据 SOP 来定义Role。\n",
    "\n",
    "**首先导入模块**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b62e394a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 17:45:23.896 | INFO     | metagpt.const:get_metagpt_package_root:29 - Package root set to /tmp/code/wow-agent/notebooks\n"
     ]
    }
   ],
   "source": [
    "import re\n",
    "\n",
    "import fire\n",
    "\n",
    "from metagpt.actions import Action, UserRequirement\n",
    "from metagpt.logs import logger\n",
    "from metagpt.roles import Role\n",
    "from metagpt.schema import Message\n",
    "from metagpt.team import Team\n",
    "\n",
    "\n",
    "def parse_code(rsp):\n",
    "    pattern = r\"```python(.*)```\"\n",
    "    match = re.search(pattern, rsp, re.DOTALL)\n",
    "    code_text = match.group(1) if match else rsp\n",
    "    return code_text"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5e98a88",
   "metadata": {},
   "source": [
    "**定义动作**\n",
    "\n",
    "我们列举了三个 Action："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "792d7348",
   "metadata": {},
   "outputs": [],
   "source": [
    "class SimpleWriteCode(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Write a python function that can {instruction}.\n",
    "    Return ```python your_code_here ```with NO other texts,\n",
    "    your code:\n",
    "    \"\"\"\n",
    "    name: str = \"SimpleWriteCode\"\n",
    "\n",
    "    async def run(self, instruction: str):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        code_text = parse_code(rsp)\n",
    "\n",
    "        return code_text\n",
    "\n",
    "class SimpleWriteTest(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Context: {context}\n",
    "    Write {k} unit tests using pytest for the given function, assuming you have imported it.\n",
    "    Return ```python your_code_here ```with NO other texts,\n",
    "    your code:\n",
    "    \"\"\"\n",
    "\n",
    "    name: str = \"SimpleWriteTest\"\n",
    "\n",
    "    async def run(self, context: str, k: int = 3):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        code_text = parse_code(rsp)\n",
    "\n",
    "        return code_text\n",
    "\n",
    "class SimpleWriteReview(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Context: {context}\n",
    "    Review the test cases and provide one critical comments:\n",
    "    \"\"\"\n",
    "\n",
    "    name: str = \"SimpleWriteReview\"\n",
    "\n",
    "    async def run(self, context: str):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(context=context)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        return rsp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f20c6ec7",
   "metadata": {},
   "source": [
    "**定义角色**\n",
    "\n",
    "在许多多智能体场景中，定义Role可能只需几行代码。对于SimpleCoder，我们做了两件事：\n",
    "\n",
    "1. 使用 set\\_actions 为Role配备适当的 Action，这与设置单智能体相同\n",
    "2. 多智能体操作逻辑：我们使Role \\_watch 来自用户或其他智能体的重要上游消息。回想我们的SOP，SimpleCoder接收用户指令，这是由MetaGPT中的UserRequirement引起的Message。因此，我们添加了 self.\\_watch([UserRequirement])。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "a667b546",
   "metadata": {},
   "outputs": [],
   "source": [
    "class SimpleCoder(Role):\n",
    "    name: str = \"Alice\"\n",
    "    profile: str = \"SimpleCoder\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self._watch([UserRequirement])\n",
    "        self.set_actions([SimpleWriteCode])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "04b7650c",
   "metadata": {},
   "source": [
    "与上述相似，对于 SimpleTester，我们：\n",
    "\n",
    "1. 使用 set\\_actions 为SimpleTester配备 SimpleWriteTest 动作\n",
    "2. 使Role \\_watch 来自其他智能体的重要上游消息。回想我们的SOP，SimpleTester从 SimpleCoder 中获取主代码，这是由 SimpleWriteCode 引起的 Message。因此，我们添加了 self.\\_watch([SimpleWriteCode])。\n",
    "\n",
    "> 一个扩展的问题：想一想如果我们使用self._watch([SimpleWriteCode,SimpleWriteReview]) 会意味着什么，可以尝试这样做\n",
    "\n",
    "此外，你可以为智能体定义自己的操作逻辑。这适用于Action需要多个输入的情况，你希望修改输入，使用特定记忆，或进行任何其他更改以反映特定逻辑的情况。因此，我们：\n",
    "\n",
    "3. 重写 \\_act 函数，就像我们在前面教程中的单智能体设置中所做的那样。在这里，我们希望SimpleTester将所有记忆用作编写测试用例的上下文，并希望有5个测试用例。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "31baf16c",
   "metadata": {},
   "outputs": [],
   "source": [
    "class SimpleTester(Role):\n",
    "    name: str = \"Bob\"\n",
    "    profile: str = \"SimpleTester\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.set_actions([SimpleWriteTest])\n",
    "        self._watch([SimpleWriteCode])\n",
    "        # self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this too\n",
    "\n",
    "    async def _act(self) -> Message:\n",
    "        logger.info(f\"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})\")\n",
    "        todo = self.rc.todo\n",
    "\n",
    "        # context = self.get_memories(k=1)[0].content # use the most recent memory as context\n",
    "        context = self.get_memories()  # use all memories as context\n",
    "\n",
    "        code_text = await todo.run(context, k=5)  # specify arguments\n",
    "        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))\n",
    "\n",
    "        return msg"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4914fd24",
   "metadata": {},
   "source": [
    "按照相同的过程定义 SimpleReviewer："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "a6af4f6d",
   "metadata": {},
   "outputs": [],
   "source": [
    "class SimpleReviewer(Role):\n",
    "    name: str = \"Charlie\"\n",
    "    profile: str = \"SimpleReviewer\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.set_actions([SimpleWriteReview])\n",
    "        self._watch([SimpleWriteTest])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3c3d11a5",
   "metadata": {},
   "source": [
    "\n",
    "**创建一个团队并添加角色**\n",
    "\n",
    "现在我们已经定义了三个 Role，是时候将它们放在一起了。我们初始化所有角色，设置一个 Team，并hire 它们。\n",
    "\n",
    "运行 Team，我们应该会看到它们之间的协作！"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f107314",
   "metadata": {},
   "outputs": [],
   "source": [
    "import fire\n",
    "import typer\n",
    "from metagpt.logs import logger\n",
    "from metagpt.team import Team\n",
    "app = typer.Typer()\n",
    "\n",
    "@app.command()\n",
    "def main(\n",
    "    idea: str = typer.Argument(..., help=\"write a function that calculates the product of a list\"),\n",
    "    investment: float = typer.Option(default=3.0, help=\"Dollar amount to invest in the AI company.\"),\n",
    "    n_round: int = typer.Option(default=5, help=\"Number of rounds for the simulation.\"),\n",
    "):\n",
    "    logger.info(idea)\n",
    "\n",
    "    team = Team()\n",
    "    team.hire(\n",
    "        [\n",
    "            SimpleCoder(),\n",
    "            SimpleTester(),\n",
    "            SimpleReviewer(),\n",
    "        ]\n",
    "    )\n",
    "\n",
    "    team.invest(investment=investment)\n",
    "    team.run_project(idea)\n",
    "    ateam.run(n_round=n_round)\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    fire.Fire(main)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "72db6c91",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
